package com.noah.common.redis.config;

import com.noah.common.redis.utility.RedisKeyUtility;
import org.redisson.Redisson;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.List;

/**
 * @author noah
 * 2023/9/7 16:45
 */
@ConditionalOnClass(Redisson.class)
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties(RedisProperties.class)
public class RedissonConfig {
    private static final String Redis_Protocol_Prefix = "redis://";
    private static final String Rediss_Protocol_Prefix = "rediss://";

    @Bean(destroyMethod = "shutdown")
    @ConditionalOnMissingBean(RedissonClient.class)
    @ConditionalOnProperty(value = "spring.data.redis.enabled")
    public RedissonClient redissonClient(RedisProperties redisProperties) {

        Config config = new Config();
        int timeout = (int) redisProperties.getTimeout().toMillis();
        int connectTimeout = (int) redisProperties.getConnectTimeout().toMillis();
        boolean isSsl = redisProperties.getSsl().isEnabled();

        if (redisProperties.getSentinel() != null) {
            config.useSentinelServers().setMasterName(redisProperties.getSentinel().getMaster())
                    .addSentinelAddress(convertNodes(isSsl, redisProperties.getSentinel().getNodes()))
                    .setDatabase(redisProperties.getDatabase())
                    .setTimeout(timeout)
                    .setConnectTimeout(connectTimeout)
                    .setPassword(redisProperties.getPassword());
        } else if (redisProperties.getCluster() != null) {
            config.useClusterServers().addNodeAddress(convertNodes(isSsl, redisProperties.getCluster().getNodes()))
                    .setPassword(redisProperties.getPassword())
                    .setTimeout(timeout)
                    .setConnectTimeout(connectTimeout);
        } else {
            config.useSingleServer().setAddress(convertAddress(isSsl, redisProperties.getHost(), redisProperties.getPort()))
                    .setDatabase(redisProperties.getDatabase())
                    .setPassword(redisProperties.getPassword())
                    .setConnectTimeout(connectTimeout)
                    .setTimeout(timeout);
        }

        //config.setExecutor(taskExecutor.getThreadPoolExecutor());

        return Redisson.create(config);
    }

    @Bean
    public RBloomFilter<String> bloomFilter(RedissonClient redissonClient) {
        RBloomFilter<String> bloomFilter = redissonClient.getBloomFilter(RedisKeyUtility.getBloomFilterKey());
        bloomFilter.tryInit(10000, 0.01);
        return bloomFilter;
    }

    private String getProtocolPrefix(boolean isSsl) {
        return isSsl ? Rediss_Protocol_Prefix : Redis_Protocol_Prefix;
    }

    /**
     * @param isSsl
     * @param host
     * @param port
     * @return
     */
    private String convertAddress(boolean isSsl, String host, int port) {
        return getProtocolPrefix(isSsl) + host + ":" + port;
    }

    private String[] convertNodes(boolean isSsl, List<String> nodeList) {
        List<String> nodes = new ArrayList<>(nodeList.size());
        for (String node : nodeList) {
            if (node.startsWith(Redis_Protocol_Prefix) || node.startsWith(Rediss_Protocol_Prefix)) {
                nodes.add(node);
            } else {
                nodes.add(getProtocolPrefix(isSsl) + node);
            }
        }
        return nodes.toArray(new String[0]);
    }
}
